package io.kimmking.rpcfx.server;
  
import com.alibaba.fastjson.JSON;
import io.kimmking.rpcfx.api.RpcfxProvider;
import io.kimmking.rpcfx.config.center.api.ConfigCenter;
import io.kimmking.rpcfx.config.center.api.ConfigInfo;
import io.kimmking.rpcfx.config.center.zookeeper.ZookeeperConfigCenter;
import io.kimmking.rpcfx.metadata.center.api.MetadataRegister;
import io.kimmking.rpcfx.metadata.center.zookeeper.ZookeeperMetadataRegister;
import io.kimmking.rpcfx.register.center.api.ProviderInfo;
import io.kimmking.rpcfx.register.center.api.Register;
import io.kimmking.rpcfx.register.center.redis.RedisRegister;
import io.kimmking.rpcfx.register.center.zookeeper.ZookeeperRegister;
import io.kimmking.rpcfx.tcp.CustomTcpDecoder;
import io.kimmking.rpcfx.tcp.CustomTcpEncoder;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;  
  
import io.netty.bootstrap.ServerBootstrap;  
import io.netty.channel.ChannelFuture;   
import io.netty.channel.ChannelInitializer;   
import io.netty.channel.ChannelOption;   
import io.netty.channel.EventLoopGroup;   
import io.netty.channel.nio.NioEventLoopGroup;   
import io.netty.channel.socket.SocketChannel;   
import io.netty.channel.socket.nio.NioServerSocketChannel;
import org.springframework.context.ApplicationContext;

import java.net.InetAddress;
import java.net.UnknownHostException;

public class NettyHttpServer {

    public void startWithRedisRegister(int port, RpcfxInvoker invoker, ApplicationContext ctx, String redisHost, int redisPort, String namespace, String cfgNamespace, String metadataNamespace) throws Exception {
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            RedisRegister register = new RedisRegister(redisHost,redisPort,namespace);
            doExport(ctx,port,invoker,null,register,null);

            ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class)
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        public void initChannel(SocketChannel ch) throws Exception {
                            ch.pipeline().addLast(new CustomTcpEncoder());
                            ch.pipeline().addLast(new CustomTcpDecoder());

                            ch.pipeline().addLast(new HttpServerInboundHandler(invoker));
                        }
                    }).option(ChannelOption.SO_BACKLOG, 128)
                    .childOption(ChannelOption.SO_KEEPALIVE, true);

            ChannelFuture f = b.bind(port).sync();
            f.channel().closeFuture().sync();
        } finally {
            workerGroup.shutdownGracefully();
            bossGroup.shutdownGracefully();
        }
    }
    public void startWithZkRegister(int port, RpcfxInvoker invoker, ApplicationContext ctx, String zkServer, String namespace, String cfgNamespace, String metadataNamespace) throws Exception {
        EventLoopGroup bossGroup = new NioEventLoopGroup();  
        EventLoopGroup workerGroup = new NioEventLoopGroup();  
        try {
            ZookeeperConfigCenter configCenter = new ZookeeperConfigCenter(zkServer,cfgNamespace);
            ZookeeperRegister register = new ZookeeperRegister(zkServer,namespace);
            ZookeeperMetadataRegister metadataRegister = new ZookeeperMetadataRegister(zkServer,metadataNamespace);
            doExport(ctx,port,invoker,configCenter,register,metadataRegister);
            configCenter.close();

            ServerBootstrap b = new ServerBootstrap();  
            b.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class)  
                    .childHandler(new ChannelInitializer<SocketChannel>() {  
                                @Override  
                                public void initChannel(SocketChannel ch) throws Exception {  
                                    ch.pipeline().addLast(new CustomTcpEncoder());
                                    ch.pipeline().addLast(new CustomTcpDecoder());

                                    ch.pipeline().addLast(new HttpServerInboundHandler(invoker));
                                }  
                            }).option(ChannelOption.SO_BACKLOG, 128)   
                    .childOption(ChannelOption.SO_KEEPALIVE, true);  
  
            ChannelFuture f = b.bind(port).sync();
            f.channel().closeFuture().sync();
        } finally {  
            workerGroup.shutdownGracefully();  
            bossGroup.shutdownGracefully();  
        }  
    }

    private void doExport(ApplicationContext ctx, int port, RpcfxInvoker invoker, ConfigCenter configCenter,
                          Register register, MetadataRegister metadataRegister ) throws UnknownHostException {
        String host = InetAddress.getLocalHost().getHostAddress();
        String[] beanNames = ctx.getBeanNamesForAnnotation(RpcfxProvider.class);
        for (String beanName : beanNames) {
            //读取各类配置
            Object bean = ctx.getBean(beanName);
            Class<?> clazz = bean.getClass();
            RpcfxProvider provider = clazz.getAnnotation(RpcfxProvider.class);

            String serverName = provider.serverName();

            String protocol = provider.protocol();

            ConfigInfo info = ConfigInfo.builder().serverName(serverName).build();

            ProviderInfo providerInfo =ProviderInfo.builder().protocol(protocol).serviceName(serverName).host(host).port(port).build();
            //读配置中心
            String config= null;
            if(configCenter !=null){
                config= configCenter.readStringConfig(info);
            }
            if(config!=null){
                //第一优先级是注解配置，第二优先级是配置中心
                //这里是配置中心
                ProviderInfo providerInfoCfg = JSON.parseObject(config,ProviderInfo.class);
                String group = providerInfoCfg.getGroup();
                String tag = providerInfoCfg.getTag();
                String version = providerInfoCfg.getVersion();
                //第一优先级是注解配置，第二优先级是配置中心
                //这里是配置中心
                if(group!=null && group.length() > 0){
                    providerInfo.setGroup(group);
                }
                if(tag !=null && tag.length() > 0){
                    providerInfo.setTag(tag);
                }
                if(version!=null && version.length() > 0){
                    providerInfo.setVersion(version);
                }
            }
            String group = provider.group();
            String tag = provider.tag();
            String version = provider.version();
            //第一优先级是注解配置，第二优先级是配置中心
            //这里是注解
            if(group.length() > 0){
                providerInfo.setGroup(group);
            }
            if(tag.length() > 0){
                providerInfo.setTag(tag);
            }
            if(version.length() > 0){
                providerInfo.setVersion(version);
            }

            //发布
            register.register(providerInfo);
            //发布metadata
            if(metadataRegister !=null){
                metadataRegister.register(serverName);

            }
        }

    }


}